package com.ds.lens.flink.monitor.forecast.function;

import com.ds.lens.data.common.constant.Constants;
import com.ds.lens.data.common.document.ForecastKafkaData;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.springframework.util.CollectionUtils;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2020-05-08
 */
public class ForecastKafkaDataFunction extends ProcessFunction<ForecastKafkaData, ForecastKafkaData> {
    private static final long serialVersionUID = 3712641548199977389L;

    @Override
    public void processElement(ForecastKafkaData forecastKafkaData, Context context, Collector<ForecastKafkaData> collector) throws Exception {
        String subTopic = forecastKafkaData.getSubTopic();
        if (StringUtils.isEmpty(subTopic) || !subTopic.contains(Constants.DASH)) {
            return;
        }
        String[] strings = subTopic.split(Constants.DASH);
        if (strings.length <= 2 || !NumberUtils.isNumber(strings[0])) {
            return;
        }
        if (forecastKafkaData == null || StringUtils.isEmpty(forecastKafkaData.getSubTopic()) || CollectionUtils.isEmpty(forecastKafkaData.getData())) {
            return;
        }
        forecastKafkaData.getData().sort((o1, o2) -> {
            if (o1.getTs() < o2.getTs()) {
                return -1;
            } else if (o1.getTs() > o2.getTs()) {
                return 1;
            } else {
                return 0;
            }
        });
        forecastKafkaData.setTimestamp(forecastKafkaData.getData().get(0).getTs());
        forecastKafkaData.setCreatedTime(System.currentTimeMillis());
        collector.collect(forecastKafkaData);
    }

}
